/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.example.simple;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.hook.ConsumeMessageContext;
import org.apache.rocketmq.client.hook.ConsumeMessageHook;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.atomic.AtomicLong;

public class PushConsumer {

    public static void main(String[] args) throws InterruptedException, Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_group");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        //订阅多个topic
        consumer.subscribe("test_topic1", "*");
        consumer.subscribe("test_topic2", "*");
        
        //consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        //wrong time format 2017_0422_221800
        //consumer.setConsumeTimestamp("20181109221800");
        
        //注册多个消费消息的Hook钩子函数
        consumer.getDefaultMQPushConsumerImpl().registerConsumeMessageHook(new ConsumeTestHook());
        
        
        AtomicLong atomicLong = new AtomicLong(System.currentTimeMillis());
        consumer.setMaxReconsumeTimes(4); //消息重试次数
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                String body = new String(msg.getBody());
                long currentTime = System.currentTimeMillis();
                long oldTime = atomicLong.get();
                atomicLong.set(currentTime);
                
                System.out.printf("topic = %s， message = %s msgId = %s time = %s %n", msg.getTopic(), body, msg.getMsgId(), currentTime - oldTime);
            }
            //return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        });
        consumer.start();
        System.out.printf("Consumer Started.%n");
        
        
        //类过滤模式：自定义类过滤（高版本内去掉了该功能）
/*        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        consumer.setNamesrvAddr("127.0.0.1:9876");
    
        ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
        File classFile = new File(classLoader.getResource("MessageFilterImpl.java").getFile());
        String filterCode = MixAll.file2String(classFile);
        consumer.subscribe("test_topic", "org.apache.rocketmq.example.filter.MessageFilterImpl", filterCode);
    
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for(MessageExt msg : msgs){
                String userId = msg.getUserProperty("userId");
                String tags = msg.getTags();
                System.out.println(tags + " userId=" + " " + userId + "  "+ new String(msg.getBody()));
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.out.println("消费者启动成功！！！");*/
    }
    
    
    /**
     * 自定义一个消费消息的钩子函数
     */
    public static class ConsumeTestHook implements ConsumeMessageHook{
        @Override
        public String hookName() {
            return ConsumeTestHook.class.getName();
        }
        
        @Override
        public void consumeMessageBefore(ConsumeMessageContext context) {
            System.out.println("execute consumeMessageBefore....");
        }
        
        @Override
        public void consumeMessageAfter(ConsumeMessageContext context) {
            System.out.println("execute consumeMessageAfter....");
        }
    }
}
